package com.jiaobuchong.zookeeper;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * @author zhoujie
 * Created on 2019-01-20 18:19
 */
public class ZkAdvancedDistributeLock implements Lock {

    /**
     * 利用临时顺序节点来实现分布式锁
     * 获取锁，取排队号（创建自己的临时顺序节点），然后判断自己是否是最小号，如果是则获取锁，不是，则注册前一节点的 watcher，
     * 阻塞等待释放锁，删除自己创建的临时顺序节点
     */
    // 父节点
    private String lockPath;

    private ZkClient client;

    // 当前取的号
    private String currentPath;

    // 前一号
    private String beforePath;

    public ZkAdvancedDistributeLock(String lockPath) {
        super();
        this.lockPath = lockPath;
        client = new ZkClient("localhost:2181");
        client.setZkSerializer(new MyZkSerializer());
        // 尝试创建父节点
        if (this.client.exists(lockPath)) {
            try {
                this.client.createPersistent(lockPath);
            } catch (ZkNodeExistsException e) {
                e.printStackTrace();
            }

        }
    }

    @Override
    public void lock() {
        // 如果获取不到锁，阻塞等待
        if (!tryLock()) {
            // 获取锁，阻塞自己
            waitForLock();
            // 再次尝试
            lock();
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    @Override
    public boolean tryLock() {
        if (this.currentPath == null) {
            // 创建临时顺序节点
            currentPath = this.client.createEphemeralSequential(lockPath + "/", "aaa");
        }
        // 获取所有子节点
        List<String> children = this.client.getChildren(lockPath);

        // 排序 List
        Collections.sort(children);

        System.out.println("children: " + children);

        // 判断当前节点是不是最小的
        if (currentPath.equals(lockPath + "/" + children.get(0))) {
            return true;
        } else {
            // 取到前一个
            // 得到字节的索引号
            int curIndex = children.indexOf(currentPath.substring(lockPath.length() + 1));
            beforePath = lockPath + "/" + children.get(curIndex - 1);
        }
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        client.delete(beforePath);
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    private void waitForLock() {
        CountDownLatch cdl = new CountDownLatch(1);
        IZkDataListener listener = new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {
                System.out.println("----收到节点数据变化：" + o + "---");
            }

            @Override
            public void handleDataDeleted(String s) throws Exception {
                System.out.println("----收到节点被删除了----");
                cdl.countDown();
            }
        };
        client.subscribeDataChanges(this.beforePath, listener);

        // 阻塞自己
        if (this.client.exists(beforePath)) {
            try {
                cdl.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 醒来后，取消 watcher
        client.unsubscribeDataChanges(this.beforePath, listener);
    }
}
